package com.lzg.okrx.observable

import com.lzy.okgo.adapter.Call
import com.lzy.okgo.callback.Callback
import com.lzy.okgo.model.Progress
import com.lzy.okgo.model.Response
import com.lzy.okgo.request.base.Request
import io.reactivex.Observable
import io.reactivex.Observer
import io.reactivex.disposables.Disposable
import io.reactivex.exceptions.CompositeException
import io.reactivex.exceptions.Exceptions
import io.reactivex.plugins.RxJavaPlugins

class CallEnqueueObservable<T>(
    private val originalCall: Call<T>
) : Observable<Response<T>>() {

    override fun subscribeActual(observer: Observer<in Response<T>>) {
        // Since Call is a one-shot type, clone it for each new observer.
        val call = originalCall.clone()
        val callback = CallCallback(call, observer)
        observer.onSubscribe(callback)
        call.execute(callback)
    }

    private class CallCallback<T> internal constructor(
        private val call: Call<T>,
        private val observer: Observer<in Response<T>>
    ) : Disposable, Callback<T> {
        internal var terminated = false

        override fun dispose() {
            call.cancel()
        }

        override fun isDisposed() = call.isCanceled

        @Throws(Throwable::class)
        override fun convertResponse(response: okhttp3.Response): T? {
            // okrx 使用converter转换，不需要这个解析方法
            return null
        }

        override fun onStart(request: Request<T, out Request<*, *>>) { }

        override fun onSuccess(response: Response<T>) {
            if (call.isCanceled) return

            try {
                observer.onNext(response)
            } catch (e: Exception) {
                if (terminated) RxJavaPlugins.onError(e)
                else onError(response)
            }
        }

        override fun onCacheSuccess(response: Response<T>) {
            onSuccess(response)
        }

        override fun onError(response: Response<T>) {
            if (call.isCanceled) return

            val throwable = response.exception
            try {
                terminated = true
                observer.onError(throwable)
            } catch (inner: Throwable) {
                Exceptions.throwIfFatal(inner)
                RxJavaPlugins.onError(CompositeException(throwable, inner))
            }
        }

        override fun onFinish() {
            if (call.isCanceled) return

            try {
                terminated = true
                observer.onComplete()
            } catch (inner: Throwable) {
                Exceptions.throwIfFatal(inner)
                RxJavaPlugins.onError(inner)
            }
        }

        override fun uploadProgress(progress: Progress) { }

        override fun downloadProgress(progress: Progress) { }
    }

}
